﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace Ingenious.Utility
{
    // Note: Loosely based upon the queued synchronization context example found at
    // http://www.codeproject.com/Articles/32113/Understanding-SynchronizationContext-Part-II

    #region Supporting Classes
    internal interface IQueueReader<T> : IDisposable
    {
        T Dequeue();
        void ReleaseReader();
    }

    internal interface IQueueWriter<T> : IDisposable
    {
        void Enqueue(T data);
    }

    internal class BlockingQueue<T> : IQueueReader<T>, IQueueWriter<T>, IDisposable
    {
        private Queue<T> m_queue = new Queue<T>();
        private Semaphore m_semaphore = new Semaphore(0, int.MaxValue);
        private ManualResetEvent m_terminateEvent = new ManualResetEvent(false);

        private WaitHandle[] m_waitHandles;

        #region Object Lifetime
        public BlockingQueue()
        {
            m_waitHandles = new WaitHandle[2] { m_semaphore, m_terminateEvent };
        }

        public void Dispose()
        {
            if (m_semaphore != null)
            {
                m_semaphore.Close();
                m_queue.Clear();
                m_semaphore = null;
            }
        }
        #endregion

        public void Enqueue(T data)
        {
            lock (m_queue)
            {
                m_queue.Enqueue(data);
            }

            m_semaphore.Release();
        }

        public T Dequeue()
        {
            // Wait until there is an item in the queue
            WaitHandle.WaitAny(m_waitHandles);

            lock (m_queue)
            {
                if (m_queue.Count > 0)
                {
                    return m_queue.Dequeue();
                }
            }

            return default(T);
        }

        public void ReleaseReader()
        {
            m_terminateEvent.Set();
        }
    }

    internal enum ExecutionType
    {
        Post,
        Send
    }

    internal class SendOrPostCallbackItem
    {
        private object m_state;
        private ExecutionType m_executionType;
        private SendOrPostCallback m_callback;
        private ManualResetEvent m_waitHandle = new ManualResetEvent(false);
        private Exception m_exception = null;

        #region Object Lifetime
        internal SendOrPostCallbackItem(SendOrPostCallback callback, object state, ExecutionType type)
        {
            if (callback == null)
            {
                throw new ArgumentNullException("callback");
            }

            m_callback = callback;
            m_state = state;
            m_executionType = type;
        }
        #endregion

        internal void Execute()
        {
            if (m_executionType == ExecutionType.Send)
            {
                Send();
            }
            else
            {
                Post();
            }
        }

        internal void Send()
        {
            try
            {
                m_callback(m_state);
            }
            catch (Exception e)
            {
                m_exception = e;
            }
            finally
            {
                m_waitHandle.Set();
            }
        }

        /// <summary />
        /// Unhandled exceptions will terminate the thread
        /// </summary />
        internal void Post()
        {
            m_callback(m_state);
        }

        #region Properties
        internal WaitHandle ExecutionCompleteWaitHandle
        {
            get { return m_waitHandle; }
        }

        internal Exception Exception
        {
            get { return m_exception; }
        }

        internal bool ExecutedWithException
        {
            get { return m_exception != null; }
        }
        #endregion
    }

    internal class ThreadFarmDispatcher : IDisposable
    {
        private Thread m_dispatcherThread;
        private IQueueReader<SendOrPostCallbackItem> m_queueConsumer;

        private ManualResetEvent m_terminateEvent = new ManualResetEvent(false);

        private int m_dispatchedThreadCount;
        private readonly object m_threadFarmLock = new object();

        private readonly int m_maxWorkerThreadCount;
        private readonly List<Thread> m_threadFarm = new List<Thread>();

        #region Object Lifetime
        internal ThreadFarmDispatcher(IQueueReader<SendOrPostCallbackItem> reader, int maxWorkerThreadCount)
        {
            if (reader == null)
            {
                throw new ArgumentNullException("reader");
            }

            if (maxWorkerThreadCount <= 0)
            {
                throw new ArgumentOutOfRangeException("maxWorkerThreadCount", "maxWorkerThreadCount must be greater than zero");
            }

            m_queueConsumer = reader;

            m_dispatcherThread = new Thread(Dispatcher);
            m_dispatcherThread.Name = "Queue Dispatch Worker";

            m_maxWorkerThreadCount = maxWorkerThreadCount;
        }

        public void Dispose()
        {
            Stop();
        }
        #endregion

        #region Threading Methods
        internal void Start()
        {
            m_dispatcherThread.Start();
        }
        
        internal void Join()
        {
            m_dispatcherThread.Join();
        }

        internal void Stop()
        {
            m_terminateEvent.Set();
            m_queueConsumer.ReleaseReader();
            m_dispatcherThread.Join();
            m_queueConsumer.Dispose();
        }
        #endregion

        #region Method Dispatcher
        private void Dispatcher()
        {
            while (true)
            {
                bool stop = m_terminateEvent.WaitOne(0);

                if (stop)
                {
                    break;
                }

                bool dispatchEvent = false;

                lock (m_threadFarmLock)
                {
                    dispatchEvent = (m_dispatchedThreadCount < m_maxWorkerThreadCount);

                    if (dispatchEvent)
                    {
                        ++m_dispatchedThreadCount;
                    }
                }

                if (dispatchEvent)
                {
                    SendOrPostCallbackItem workItem = m_queueConsumer.Dequeue();

                    Thread thread = new Thread((o) =>
                    {
                        if (workItem != null)
                        {
                            workItem.Execute();
                        }

                        lock (m_threadFarmLock)
                        {
                            --m_dispatchedThreadCount;
                        }
                    });

                    lock (m_threadFarmLock)
                    {
                        m_threadFarm.Add(thread);
                        thread.Start();
                    }
                }

                // Sanitize the list of spent threads
                lock (m_threadFarmLock)
                {
                    HashSet<Thread> deadThreads = new HashSet<Thread>(m_threadFarm.Where((t) => !t.IsAlive));
                    m_threadFarm.RemoveAll((t) => deadThreads.Contains(t));
                }
            }
        }
        #endregion
    }
    #endregion

    /// <summary>
    /// Special synchronization context that queues items and dispatches tasks to a
    /// configurable number of concurrent thread pool threads
    /// </summary>
    public class DispatcherSynchronizationContext : SynchronizationContext, IDisposable
    {
        private BlockingQueue<SendOrPostCallbackItem> m_queue;
        private ThreadFarmDispatcher m_threadFarmDispatcher;

        public DispatcherSynchronizationContext(int maxWorkerThreadCount)
            : base()
        {
            if (maxWorkerThreadCount <= 0)
            {
                throw new ArgumentOutOfRangeException("maxWorkerThreadCount", "maxWorkerThreadCount must be greater than zero");
            }

            m_queue = new BlockingQueue<SendOrPostCallbackItem>();

            m_threadFarmDispatcher = new ThreadFarmDispatcher(m_queue, maxWorkerThreadCount);
            m_threadFarmDispatcher.Start();
        }

        public override void Send(SendOrPostCallback d, object state)
        {
            // Create an item for execution
            SendOrPostCallbackItem item = new SendOrPostCallbackItem(d, state, ExecutionType.Send);

            // Queue the item
            m_queue.Enqueue(item);

            // Wait for the item execution to end (blocking call)
            item.ExecutionCompleteWaitHandle.WaitOne();

            // If there was an exception, rethrow on the caller thread
            if (item.ExecutedWithException)
            {
                throw item.Exception;
            }
        }

        public override void Post(SendOrPostCallback d, object state)
        {
            // Queue the item and don't wait for its execution. This is risky because
            // an unhandled exception will terminate the farmed thread. Use with caution.
            SendOrPostCallbackItem item = new SendOrPostCallbackItem(d, state, ExecutionType.Post);
            m_queue.Enqueue(item);
        }

        public void Dispose()
        {
            m_threadFarmDispatcher.Dispose();
        }

        public override SynchronizationContext CreateCopy()
        {
            return this;
        }
    }
}
